package com.demo.rabbitmq.demo02;

import com.demo.rabbitmq.contanst.Constant;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class WorkQueueReveive {

    public static void execute(String host, String username, String password, int id) {
        // 配置连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setUsername(username);
        factory.setPassword(password);

        Connection connection = null;
        Channel channel = null;

        try {
            // 建立TCP连接
            connection = factory.newConnection();
            // 在TCP连接基础上创建信道
            channel = connection.createChannel();
            // 声明一个队列
            channel.queueDeclare(Constant.TASK_QUEUE_NAME, true, false, false, null);
            System.out.println("[WorkQueueReceive-" + id + "] Waiting For Messages");

            // 每个客户端每次最后获取N个消息
            channel.basicQos(1);

            // 默认消费者实现
            Channel finalChannel = channel;
            Consumer consumer = new DefaultConsumer(finalChannel) {
                final int mID = id;

                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("[WorkQueueReceive-" + mID + "] Received : " + message);
                    System.out.println("[WorkQueueReceive-" + mID + "] Done...");

                    // 情况一： 对处理好的信息进行应答
                    // finalChannel.basicAck(envelope.getDeliveryTag(), false);

                    // 情况二：对于id=0 的消息正常应答，其他，拒绝此消息并要求重发
                    if (mID == 1) {
                        // 对处理好的消息进行应答
                        finalChannel.basicAck(envelope.getDeliveryTag(), false);
                        System.out.println("Receiver Response " + mID + "-- DeliveryTag: " + envelope.getDeliveryTag());
                    } else {
                        // 拒绝当前这条消息
                        System.out.println("Receiver Reject " + mID + "-- DeliveryTag: " + envelope.getDeliveryTag());
                        finalChannel.basicReject(envelope.getDeliveryTag(), true);
                        // 拒绝包含本条 delivery_tag 所对应消息在内的所有比该值小的消息都被拒绝了（除了被ACK的意外）
                        finalChannel.basicNack(envelope.getDeliveryTag(), false, false);
                    }
                }
            };

            // 获取消息
            channel.basicConsume(Constant.TASK_QUEUE_NAME, false, consumer);

        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}
